package kafkareport

import (
	"context"
	"dataflow-operator/k8s"
	"encoding/json"
	"os"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/go-logr/logr"

	dfv1 "dataflow-operator/api/v1"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/selection"
	cmdutil "k8s.io/kubectl/pkg/cmd/util"
	metricsclientset "k8s.io/metrics/pkg/client/clientset/versioned"
)

type KafkaReporter struct {
	KafkaProducerConfig *sarama.Config
	producer            sarama.SyncProducer
	Factory             cmdutil.Factory
	Logger              logr.Logger
}

func (kr *KafkaReporter) Start(ctx context.Context) {
	var err error
	address := strings.Split(Config.Brokers, ",")
	kr.producer, err = sarama.NewSyncProducer(address, kr.KafkaProducerConfig)
	if err != nil {
		kr.Logger.Error(err, "Parse kafka config error")
		os.Exit(-1)
	}

	config, err := kr.Factory.ToRESTConfig()
	if err != nil {
		kr.Logger.Error(err, "Get k8s rest config error")
		os.Exit(-1)
	}

	metricsClient, err := metricsclientset.NewForConfig(config)
	clientset, err := kr.Factory.KubernetesClientSet()
	if err != nil {
		kr.Logger.Error(err, "Get k8s client error")
		os.Exit(-1)

	}
	podClient := clientset.CoreV1().Pods(dfv1.DataflowNamespace)

	selector := labels.Everything()
	requrment, err := labels.NewRequirement(dfv1.DataflowPodLabelKey, selection.In, dfv1.DataflowPodLabelVals)
	if err != nil {
		kr.Logger.Error(err, "Create labels error")
		os.Exit(-1)
	}
	selector = selector.Add(*requrment)

	for {
		select {
		case <-ctx.Done():
			kr.Logger.Info("Stop kafka reporter.")
		default:
			podList, err := podClient.List(context.Background(), metav1.ListOptions{
				LabelSelector: selector.String(),
			})
			if err != nil {
				kr.Logger.Error(err, "Kafka report get pod failed")
				os.Exit(-1)
			}
			metricsList, err := k8s.GetMetricsFromMetricsAPI(metricsClient, dfv1.DataflowNamespace, "", false, selector)
			appList, err := k8s.CalculatePodMetrics(podList, metricsList)
			if err != nil {
				kr.Logger.Error(err, "CalculatePodMetrics error")
				continue
			}
			bytes, err := json.Marshal(appList)
			if err != nil {
				kr.Logger.Error(err, "Marshal App list error")
				continue
			}
			msg := &sarama.ProducerMessage{
				Topic: Config.Topic,
				Value: sarama.ByteEncoder(bytes),
			}
			_, _, err = kr.producer.SendMessage(msg)
			if err != nil {
				kr.Logger.Error(err, "Kafka send error")
				continue
			}

			time.Sleep(5 * time.Second)
		}
	}
}
